package com.bytes32.classification.tree.dataset;

import java.io.ByteArrayOutputStream;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.io.WritableUtils;

/**
 * Used for labels, the input
 * value is distilled into a byte array
 * to allow for various labels but String
 * should be the main type.
 * @author murariuf
 *
 */
public final class UnOrderedProcessor implements FeatureProcessor<Entry<ByteArrayWrapper, Long>>{

	private Map<ByteArrayWrapper, Long> counters = new HashMap<ByteArrayWrapper, Long>();
	
	private long totalCount;
	
	@Override
	public void readFields(DataInput stream) throws IOException {
		int size = WritableUtils.readVInt(stream);
		for (int i = 0; i< size; i++){
			ByteArrayWrapper key = new ByteArrayWrapper();
			key.readFields(stream);
			Long count = WritableUtils.readVLong(stream);
			counters.put(key, count);
		}
	}

	@Override
	public void write(DataOutput stream) throws IOException {
		WritableUtils.writeVInt(stream, counters.size());
		for (Entry<ByteArrayWrapper, Long> count : counters.entrySet()){
			count.getKey().write(stream);
			WritableUtils.writeVLong(stream, count.getValue());
		}
	}

	@Override
	public long getCount() {
		return counters.size();
	}

	@Override
	public void handleString(String seq) throws InvalidFeatureValueException {
		add(seq.getBytes());
	}
	
	private void add(byte[] value){
		ByteArrayWrapper wrapper = new ByteArrayWrapper();
		wrapper.setWrapped(value);
		Long counter = this.counters.get(wrapper);
		counter = counter != null ? counter += 1L : 1;
		this.counters.put(wrapper, counter);
		totalCount+=1;
	}

	@Override
	public Iterator<Entry<ByteArrayWrapper, Long>> iterator() {
		return Collections.unmodifiableSet(this.counters.entrySet()).iterator();
	}

	public long getTotalCount() {
		return totalCount;
	}
	
	public UnOrderedProcessor merge(Iterator<UnOrderedProcessor> buffer){
		while (buffer.hasNext()){
			UnOrderedProcessor temp = buffer.next();
			Iterator<Entry<ByteArrayWrapper, Long>> iter = temp.iterator();
			while (iter.hasNext()){
				Entry<ByteArrayWrapper, Long> counter = iter.next();
				Long count = this.counters.get(counter.getKey());
				if (count == null){
					this.counters.put(counter.getKey(), counter.getValue());
				} else {
					count+=counter.getValue();
					this.counters.put(counter.getKey(), count);
				}
			}
		}
		return this;
	}
	
	@Override
	public String toString() {
		return this.counters.toString();
	}
	
	@Override
	public void compact() {
		/* Nothing */
	}
	
}
